package polimi.reds.optimizer;

import java.io.ByteArrayOutputStream;



import java.io.FileWriter;
import java.io.IOException;
import java.io.ObjectOutputStream;
import java.io.Serializable;

import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;

import polimi.reds.Filter;
import polimi.reds.Message;
import polimi.reds.NodeDescriptor;
import polimi.reds.broker.overlay.NeighborhoodChangeListener;
import polimi.reds.broker.overlay.Overlay;
import polimi.reds.broker.overlay.PacketListener;
import polimi.reds.broker.routing.Router;
import polimi.reds.broker.routing.SubscriptionTable;



/**
 * This abstract class represents a basic reusable implementation of OverlayOptimizer.
 * <p>
 * The functionalities provided by this class include methods for issuing broker swaps
 * in the neighborhood of the overlay.
 * 
 * @author Daniel Dubois
 */
public abstract class AbstractOptimizerStrategy implements OptimizerStrategy {
		
	protected Logger logger = Logger.getLogger("polimi.reds.optimizer.OverlayOptimizer");
	private Router router=null;
	private Overlay overlay=null;
	private SubscriptionTable subscriptionTable;
	private Map<String,Double> stats=new HashMap<String,Double>();
	private SwapStrategy swapThread;
	
	public void initialize(Overlay overlay, Router router, Map<String,String> options, TrafficLock trafficLock) {
		
		try {
			FileWriter file = new FileWriter(overlay.getNodeDescriptor().getID()+".txt",false);
			file.append(new Date() + " Starting simulation on site: " + overlay.getNodeDescriptor().getID() + "\n");
			file.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
		
		setOverlay(overlay);
		setRouter(router);
		
		for (String s:options.keySet()) {
			addToTrace("OPTION " + s + "=" + options.get(s));
		}
		swapThread = new SwapStrategy(this, trafficLock, Integer.parseInt(options.get(PARAM_MAXNEIGHBORS)), router);
		swapThread.setPriority(Thread.MAX_PRIORITY);
		swapThread.setDaemon(true);
	}
	
	protected abstract void preRoutingTrafficEvent(NodeDescriptor source, Message data);
	protected abstract void postRoutingTrafficEvent(NodeDescriptor source, NodeDescriptor destination, Message data);
	protected abstract void subscriptionTableUpdateEvent(SubscriptionTable subscriptionTable);
	protected abstract void brokersUpdateEvent(Collection<NodeDescriptor> nodes);
	
	protected synchronized final void tryToSwapNeighbor(NodeDescriptor executorNode, NodeDescriptor nodeToRemove, NodeDescriptor nodeToAdd) {
		
		logger.finest("tryToSwapNeighbor: " + overlay.getNodeDescriptor().getID() + "@" + executorNode.getID() + " -" + nodeToRemove.getID() + " +" + nodeToAdd.getID());
		swapThread.beginSwapSession(nodeToRemove, executorNode, nodeToAdd, (new Random().nextInt()));
	}

	private void setOverlay(Overlay overlay) {
		if (this.overlay != overlay) {
			MyEventAdapter myEventAdapter = new MyEventAdapter();
			overlay.addPacketListener(myEventAdapter, OptimizerStrategy.REMOTE_SWAP_N1);
			overlay.addPacketListener(myEventAdapter, OptimizerStrategy.REMOTE_SWAP_N2);
			overlay.addPacketListener(myEventAdapter, OptimizerStrategy.REMOTE_ACK);
			overlay.addPacketListener(myEventAdapter, Router.PUBLISH);
			overlay.addPacketListener(myEventAdapter, Router.SUBSCRIBE);
			overlay.addPacketListener(myEventAdapter, Router.UNSUBSCRIBE);
			overlay.addPacketListener(myEventAdapter, Router.UNSUBSCRIBEALL);
			overlay.addNeighborhoodChangeListener(myEventAdapter);
			this.overlay=overlay;
			brokersUpdateEvent(overlay.getNeighbors());
		}
	}
	
	protected Overlay getOverlay() {
		return overlay;
	}

	private void setRouter(Router router) {
		if (this.router != router) {
			this.router=router;
			this.subscriptionTable=router.getSubscriptionTable();
			subscriptionTableUpdateEvent(this.subscriptionTable);
		}
	}
	
	@SuppressWarnings("unchecked")
	private synchronized void generatePostRoutingTrafficEvents(NodeDescriptor source,
			Message data) {
		
		Collection<NodeDescriptor> destinations = subscriptionTable.matches(data,source);
		for (NodeDescriptor n:destinations) {
			
			if (!n.isBroker()) {
				Double size = new Double(packetSize(source,n,data));
				addStat(STAT_PUB_OUTGOING_TRAFFIC_EVENTS,1.0);
				addStat(STAT_PUB_OUTGOING_TRAFFIC_SIZE,size);
			}
			
			if (overlay.isNeighborOf(n)) {
				addToTrace(Router.PUBLISH + " TO " + n.getID() + " [" + data + "]");
				postRoutingTrafficEvent(source, n, data);
			}
		}
	}
	
	private class MyEventAdapter implements PacketListener, NeighborhoodChangeListener {

		@Override
		public synchronized void notifyPacketArrived(String subject, NodeDescriptor source,
				Serializable packet) {
			
			// update statistics
			Double size = new Double(packetSize(subject,source,packet));
				
			if (source.isBroker()) {
				addStat(STAT_TOTAL_OVERLAY_TRAFFIC_EVENTS,1.0);
				addStat(STAT_TOTAL_OVERLAY_TRAFFIC_SIZE,size);
				if (subject.equals(Router.SUBSCRIBE) || subject.equals(Router.UNSUBSCRIBE) || subject.equals(Router.UNSUBSCRIBEALL)) {
					addStat(STAT_SUB_OVERLAY_TRAFFIC_EVENTS,1.0);
					addStat(STAT_SUB_OVERLAY_TRAFFIC_SIZE,size);
				} else if (subject.equals(Router.PUBLISH)) {
					addStat(STAT_PUB_OVERLAY_TRAFFIC_EVENTS,1.0);
					addStat(STAT_PUB_OVERLAY_TRAFFIC_SIZE,size);
				}
			} else {
				addStat(STAT_TOTAL_INCOMING_TRAFFIC_EVENTS,1.0);
				addStat(STAT_TOTAL_INCOMING_TRAFFIC_SIZE,size);
				if (subject.equals(Router.SUBSCRIBE) || subject.equals(Router.UNSUBSCRIBE) || subject.equals(Router.UNSUBSCRIBEALL)) {
					addStat(STAT_SUB_INCOMING_TRAFFIC_EVENTS,1.0);
					addStat(STAT_SUB_INCOMING_TRAFFIC_SIZE,size);
				} else if (subject.equals(Router.PUBLISH)) {
					addStat(STAT_PUB_INCOMING_TRAFFIC_EVENTS,1.0);
					addStat(STAT_PUB_INCOMING_TRAFFIC_SIZE,size);
				}
			}
				
			// perform actions
				
			if (subject.equals(Router.SUBSCRIBE) || subject.equals(Router.UNSUBSCRIBE) || 
					subject.equals(Router.UNSUBSCRIBEALL)) {
				logger.fine(subject + " message received in " + overlay.getNodeDescriptor().getID() +
						" from " + source.getID() + " [" + packet + "]");
				addToTrace(subject + " FROM " + source.getID() + " [" + packet + "]");
				subscriptionTableUpdateEvent(router.getSubscriptionTable());
				swapThread.notifyPubSubEvent();
			} else if (subject.equals(Router.PUBLISH)) {
				logger.fine(subject + " message received in " + overlay.getNodeDescriptor().getID() +
						" from " + source.getID() + " [" + packet + "]");
				preRoutingTrafficEvent(source, (Message) packet);
				addToTrace(subject + " FROM " + source.getID() + " [" + packet + "]");
				generatePostRoutingTrafficEvents(source, (Message) packet);
				swapThread.notifyPubSubEvent();
			} else if (subject.equals(Router.REPLY)) {
				logger.warning("Unhandled " + subject + " message received in " + overlay.getNodeDescriptor().getID() +
						" from " + source.getID() + " [" + packet + "]");
			} else if (subject.equals(OptimizerStrategy.REMOTE_SWAP_N1)) {
				Serializable params[] = (Serializable[]) packet;
				swapThread.beginSwapSession(source, overlay.getNodeDescriptor(), (NodeDescriptor) params[0], (Integer) params[1]);
			} else if (subject.equals(OptimizerStrategy.REMOTE_SWAP_N2)) {
				Serializable params[] = (Serializable[]) packet;
				swapThread.beginSwapSession(source, (NodeDescriptor) params[0], overlay.getNodeDescriptor(), (Integer) params[1]);
			} else if (subject.equals(OptimizerStrategy.REMOTE_ACK)) {
				Serializable params[] = (Serializable[]) packet;
				if (params.length==2) {
					swapThread.notifyAck((Integer) params[0], (Integer) params[1], source, null);
				} else if (params.length==3) {
					swapThread.notifyAck((Integer) params[0], (Integer) params[1], source, params[2]);
				}
				logger.fine(subject + " message received in " + overlay.getNodeDescriptor().getID() +
						" from " + source.getID() + " [" + packet + "]");
			}
		}
		
		public void notifyNeighborAdded(NodeDescriptor addedNeighbor,
				Serializable reconfInfo) {
			
			if (addedNeighbor.isBroker()) {
				logger.fine("Node " + overlay.getNodeDescriptor().getID() + " receives new broker neighbor " + 
						addedNeighbor.getID());
				addToTrace("+BROKER " + addedNeighbor.getID());
				//addToTrace("MyTable is now: " + subscriptionTable);
				brokersUpdateEvent(overlay.getNeighbors());
			} else {
				logger.fine("Node " + overlay.getNodeDescriptor().getID() + " receives new non-broker neighbor " + 
						addedNeighbor.getID());
			}
			
		}

		@Override
		public void notifyNeighborDead(NodeDescriptor deadNeighbor,
				Serializable reconfInfo) {
			if (deadNeighbor.isBroker()) {
				logger.fine("Node " + overlay.getNodeDescriptor().getID() + " loses broker neighbor " + 
						deadNeighbor.getID() + " (link dead)");
				addToTrace("-BROKER " + deadNeighbor.getID() + " (dead neighbor)");
				//addToTrace("MyTable is now: " + subscriptionTable);
				brokersUpdateEvent(overlay.getNeighbors());
			} else {
				logger.fine("Node " + overlay.getNodeDescriptor().getID() + " loses non-broker neighbor " + 
						deadNeighbor.getID() + " (link dead)");
			}
		}

		@Override
		public void notifyNeighborRemoved(NodeDescriptor removedNeighbor) {
			if (removedNeighbor.isBroker()) {
				logger.fine("Node " + overlay.getNodeDescriptor().getID() + " loses broker neighbor " + 
						removedNeighbor.getID() + " (link removed)");
				addToTrace("-BROKER " + removedNeighbor.getID() + " (clean removal)");
				//addToTrace("MyTable is now: " + subscriptionTable);
				brokersUpdateEvent(overlay.getNeighbors());
			} else {
				logger.fine("Node " + overlay.getNodeDescriptor().getID() + " loses non-broker neighbor " + 
						removedNeighbor.getID() + " (link removed)");
			}
		}
	}
	
	protected void addStat(String name, Double value) {
		
		synchronized (stats) {
			Double stat = getStat(name);
		
			if (stat == null) {
				stat = 0.0;
			}
		
			stat += value;
			setStat(name,stat);
		}
	}
	
	protected void setStat(String name, Double value) {
		
		synchronized (stats) {
			stats.put(name, value);
		}
	}
	
	protected Double getStat(String name) {
		synchronized (stats) {
			return stats.get(name);
		}
	}
	
	public Map<String,Double> getStats() {
		
		synchronized (stats) {
			return new HashMap<String,Double>(stats);
		}
	}

	protected int packetSize(Serializable ... packets) {
		int size=0;
		
		try {
			ByteArrayOutputStream bOut = new ByteArrayOutputStream();  
			ObjectOutputStream oOut = new ObjectOutputStream(bOut);  
			oOut.writeObject(packets);  
			oOut.close();
			size += bOut.toByteArray().length;
		} catch (IOException e) {
		}
		return size;
	}
	
	public void addToTrace(String message) {
		
		try {
			FileWriter file = new FileWriter(overlay.getNodeDescriptor().getID()+".txt",true);
			file.append(new Date() + " " + message + "\n");
			file.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
}
